package com.pig4cloud.pigx.strategy.executor.contract.stop_order.impl;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.okex.open.api.bean.trade.param.CancelAlgoOrder;
import com.okex.open.api.bean.trade.param.PlaceAlgoOrder;
import com.okex.open.api.config.APIConfiguration;
import com.okex.open.api.customize.constant.ApiOkxConstant;
import com.okex.open.api.customize.enums.ApiOkxEnum;
import com.okex.open.api.customize.util.ApiOkxUtil;
import com.okex.open.api.exception.APIException;
import com.okex.open.api.service.trade.TradeAPIService;
import com.okex.open.api.service.trade.impl.TradeAPIServiceImpl;
import com.pig4cloud.pigx.common.customize.enums.CommonDictEnum;
import com.pig4cloud.pigx.common.customize.exception.ApiRequestException;
import com.pig4cloud.pigx.common.customize.exception.DuckweedException;
import com.pig4cloud.pigx.common.customize.exception.MqAckException;
import com.pig4cloud.pigx.common.customize.exception.MqAckLogException;
import com.pig4cloud.pigx.common.customize.exception.MqNotAckException;
import com.pig4cloud.pigx.common.customize.mysql.utils.MysqlUtil;
import com.pig4cloud.pigx.common.customize.redis.constants.RedisKeyEnum;
import com.pig4cloud.pigx.common.customize.redis.utils.RedisUtils;
import com.pig4cloud.pigx.common.customize.utils.BigDecimalUtils;
import com.pig4cloud.pigx.common.customize.utils.CommonUtil;
import com.pig4cloud.pigx.strategy.bo.StrategyStopOrderContractCacheBO;
import com.pig4cloud.pigx.strategy.bo.StrategyTaskContractCacheBO;
import com.pig4cloud.pigx.strategy.bo.okx.PlaceOrderBO;
import com.pig4cloud.pigx.strategy.bo.okx.PlaceStopOrderBO;
import com.pig4cloud.pigx.strategy.bo.okx.SyncStopOrderBO;
import com.pig4cloud.pigx.strategy.dto.StrategyStopOrderContractCancelBatchDTO;
import com.pig4cloud.pigx.strategy.dto.StrategySyncOrderMsgDTO;
import com.pig4cloud.pigx.strategy.dto.StrategySyncStopOrderMsgDTO;
import com.pig4cloud.pigx.strategy.entity.StrategyContract;
import com.pig4cloud.pigx.strategy.entity.StrategyPositionContract;
import com.pig4cloud.pigx.strategy.entity.StrategyStopOrderContract;
import com.pig4cloud.pigx.strategy.entity.StrategyTaskContract;
import com.pig4cloud.pigx.strategy.enums.StrategyOrderContractEnum;
import com.pig4cloud.pigx.strategy.enums.StrategyPositionContractEnum;
import com.pig4cloud.pigx.strategy.enums.StrategyStopOrderContractEnum;
import com.pig4cloud.pigx.strategy.enums.StrategyTaskContractEnum;
import com.pig4cloud.pigx.strategy.executor.contract.stop_order.StrategyStopOrderContractHandler;
import com.pig4cloud.pigx.strategy.executor.contract.stop_order.StrategyStopOrderContractHandlerManager;
import com.pig4cloud.pigx.strategy.executor.contract.task.StrategyTaskContractHandler;
import com.pig4cloud.pigx.strategy.mapper.StrategyOrderContractMapper;
import com.pig4cloud.pigx.strategy.mapper.StrategyStopOrderContractMapper;
import com.pig4cloud.pigx.strategy.mapper.StrategyTaskContractMapper;
import com.pig4cloud.pigx.strategy.util.StrategyContractUtil;
import com.pig4cloud.pigx.system.constant.SystemPlatformConstant;
import com.pig4cloud.pigx.system.entity.SystemProduct;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * 欧易订单处理器
 *
 * @since 2022/11/29 17:22
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class StrategyStopOrderContractOKXHandler implements StrategyStopOrderContractHandler {

    public static final Long PLATFORM_ID = SystemPlatformConstant.OKX;
    public static final ThreadLocal<APIConfiguration> API_CONFIG = new ThreadLocal<>();
    public static final ThreadLocal<TradeAPIService> TRADE_CLIENT = new ThreadLocal<>();

    private final StrategyTaskContractHandler strategyTaskContractHandler;

    private final StrategyOrderContractMapper strategyOrderContractMapper;
    private final StrategyStopOrderContractMapper strategyStopOrderContractMapper;
    private final StrategyTaskContractMapper strategyTaskContractMapper;
    private final StrategyContractUtil strategyContractUtil;
    private final MysqlUtil mysqlUtil;
    private final RedisUtils redisUtils;

    /**
     * 注册处理器
     *
     * @since 2022/12/3 21:12
     */
    @Override
    public void afterPropertiesSet() {
        StrategyStopOrderContractHandlerManager.HANDLER_LIST.add(this);
    }

    /**
     * 根据平台id判断是否支持
     *
     * @param platformId 平台id
     * @return {@link boolean}
     * @since 2022/11/29 17:27
     */
    @Override
    public boolean support(Long platformId) {
        return PLATFORM_ID.equals(platformId);
    }

    /**
     * 获取用户apiKey信息
     *
     * @param userId 用户id
     * @return {@link APIConfiguration}
     * @since 2023/2/10 09:51
     */
    private APIConfiguration getApiConfig(Long userId) throws ApiRequestException {
        if (Objects.isNull(API_CONFIG.get())) {
            APIConfiguration apiConfig = strategyContractUtil.getOkxApiConfig(userId, PLATFORM_ID);
            API_CONFIG.set(apiConfig);
            return apiConfig;
        }

        return API_CONFIG.get();
    }

    /**
     * 根据用户id初始化请求参数
     *
     * @param userId 用户id
     * @since 2022/11/30 17:12
     */
    public void initTradeClient(Long userId) throws ApiRequestException {
        APIConfiguration apiConfig = getApiConfig(userId);
        // 交易连接设置为不重试
        apiConfig.setRetryOnConnectionFailure(false);
        TradeAPIService tradeClient = new TradeAPIServiceImpl(apiConfig);
        TRADE_CLIENT.set(tradeClient);
    }

    /**
     * 累计同步止盈止损次数
     *
     * @param id id
     * @return {@link Long}
     * @since 2022/12/2 15:00
     */
    private Long incrSyncStopOrderCount(Long id) {
        return redisUtils.addIncrCache(RedisKeyEnum.STRATEGY_STOP_ORDER_CONTRACT_SYNC_COUNT, id, 1);
    }

    /**
     * 创建止盈止损委托
     *
     * @param bo bo
     * @since 2023/2/8 13:46
     */
    @Override
    public void place(PlaceStopOrderBO bo) {
        try {
            // 获取代理商代码
            String okxBrokerCode = strategyContractUtil.getOKXBrokerCode();
            bo.setTag("okxBrokerCode：" + okxBrokerCode);

            StrategyTaskContractCacheBO taskCache = bo.getTaskCache();
            if (taskCache.getPendingOrderNum() > 0) {
                throw new MqAckException("当前任务存在挂单，无法设置止盈止损");
            }

            // 初始化客户端
            initTradeClient(bo.getUserId());
            // 开启本地事务
            mysqlUtil.startLocalTx();

            // 修改仓位
            placeHandlePosition(bo);

            // 组装下单参数
            PlaceAlgoOrder placeAlgoOrder = new PlaceAlgoOrder();
            placeAlgoOrder.setInstId(bo.getProductName());
            placeAlgoOrder.setTdMode(ApiOkxEnum.PositionMode.CROSS.getName());
            placeAlgoOrder.setSide(ApiOkxUtil.getSide(bo.getSide(), StrategyOrderContractEnum.OperationType.CLOSE.getValue()));
            placeAlgoOrder.setPosSide(ApiOkxEnum.Side.findName(bo.getSide()));
            placeAlgoOrder.setOrdType(ApiOkxEnum.AlgoOrderType.getBySideType(bo.getSideType()));
            placeAlgoOrder.setSz(bo.getNum().toString());
            placeAlgoOrder.setTag(okxBrokerCode);
            placeAlgoOrder.setClOrdId(bo.getId().toString());

            // 设置了止盈
            if (Objects.equals(CommonDictEnum.YesOrNo.YES.getValue(), bo.getHasStopProfit())) {
                placeAlgoOrder.setTpTriggerPx(BigDecimalUtils.toStrRemoveExtraZero(bo.getStopProfitTrigger()));
                if (Objects.equals(StrategyStopOrderContractEnum.StopType.LIMIT.getValue(), bo.getStopProfitType())) {
                    // 限价
                    placeAlgoOrder.setTpOrdPx(BigDecimalUtils.toStrRemoveExtraZero(bo.getStopProfitPrice()));
                } else {
                    // 市价
                    placeAlgoOrder.setTpOrdPx("-1");
                }
            }

            // 设置了止损
            if (Objects.equals(CommonDictEnum.YesOrNo.YES.getValue(), bo.getHasStopLost())) {
                placeAlgoOrder.setSlTriggerPx(BigDecimalUtils.toStrRemoveExtraZero(bo.getStopLostTrigger()));
                // 限价
                if (Objects.equals(StrategyStopOrderContractEnum.StopType.LIMIT.getValue(), bo.getStopLostType())) {
                    placeAlgoOrder.setSlOrdPx(BigDecimalUtils.toStrRemoveExtraZero(bo.getStopLostPrice()));
                } else {
                    // 市价
                    placeAlgoOrder.setSlOrdPx("-1");
                }
            }

            // 下单
            JSONObject res = TRADE_CLIENT.get().placeAlgoOrder(placeAlgoOrder);
            boolean success = ApiOkxUtil.isSuccess(res, false);
            if (!success) {
                throw new MqAckLogException("ok接口异常，报错信息：" + ApiOkxUtil.getResMsg(res));
            }

            bo.setExternalId(getPlaceStopOrderExternalId(res));
            // 新增止盈止损记录
            if (strategyStopOrderContractMapper.insert(bo) != 1) {
                throw new MqAckLogException("新增止盈止损记录失败");
            }

            // 更新任务缓存
            taskCache.addStopOrderCache(bo);
            redisUtils.flushPersistentCache(RedisKeyEnum.STRATEGY_TASK_CONTRACT, taskCache.getId(), taskCache);
            // 提交本地事务
            mysqlUtil.commitLocalTx();
        } catch (Throwable e) {
            handlePlaceError(bo, e);
            throw new MqAckLogException(e.getMessage());
        }
    }

    /**
     * 处理设置止盈止损发生异常
     *
     * @param res res
     * @return {@link Long}
     * @since 2023/2/8 19:54
     */
    private static Long getPlaceStopOrderExternalId(JSONObject res) {
        JSONArray data = res.getJSONArray(ApiOkxConstant.DATA);
        if (data.isEmpty()) {
            throw new DuckweedException("ok返回结果参数异常");
        }

        JSONObject item = data.getJSONObject(0);
        if (Objects.isNull(item)) {
            throw new DuckweedException("ok返回结果参数异常");
        }

        Long externalId = item.getLong("algoId");
        if (Objects.isNull(externalId)) {
            throw new DuckweedException("ok返回结果参数异常");
        }

        return externalId;
    }

    /**
     * 保存错误止盈止损
     *
     * @since 2022/12/11 15:07
     */
    public void handlePlaceError(PlaceStopOrderBO bo, Throwable e) {
        // 打印日志
        strategyContractUtil.logApiErrorMsg("placeStopOrder", e, bo);
        // 回滚本地事务
        mysqlUtil.rollbackLocalTx();

        if (Objects.isNull(bo.getNum())) {
            bo.setNum(0);
        }
        bo.setRemark(e.getMessage());
        bo.setStatus(e instanceof APIException ?
                StrategyStopOrderContractEnum.Status.HTTP_ERROR.getValue()
                : StrategyStopOrderContractEnum.Status.ERROR.getValue());

        // 新增订单记录
        if (strategyStopOrderContractMapper.insert(bo) != 1) {
            throw new MqAckLogException("新增异常止盈止损记录失败");
        }
    }

    /**
     * 下单时修改仓位
     *
     * @param bo bo
     * @since 2022/12/6 18:26
     */
    private void placeHandlePosition(PlaceStopOrderBO bo) {
        StrategyTaskContractCacheBO taskCache = bo.getTaskCache();
        // 查询仓位列表，并锁定
        List<StrategyPositionContract> positionList = strategyContractUtil.getPositionList(bo.getTaskId());
        bo.setPositionList(positionList);
        taskCache.setPositionList(positionList);
        // 获取仓位信息
        StrategyPositionContract position = StrategyContractUtil.getCurrentPositionBySide(positionList, bo.getSide(), false);
        bo.setPositionId(position.getId());

        if (!Objects.equals(StrategyPositionContractEnum.Status.HOLDING.getValue(), position.getStatus())) {
            throw new MqAckException("仓位状态不是持仓中，无法设置止盈止损");
        }

        // 数量类型 = 张数
        int canStopNum = position.getNumNow() - position.getNumStopOrder();
        // 数量类型 = 百分比，只能传入100%
        if (Objects.equals(StrategyStopOrderContractEnum.NumType.PERCENT.getValue(), bo.getNumType())) {
            bo.setNum(canStopNum);
            if (bo.getNum() == 0) {
                throw new MqAckException("当前仓位没有可以止盈止损的持仓数量");
            }
        }

        if (bo.getNum() > canStopNum) {
            throw new MqAckException("止盈止损数量不能超过可止盈数量：" + canStopNum);
        }

        position.setNumStopOrder(position.getNumStopOrder() + bo.getNum());
        position.setUpdateTime(LocalDateTime.now());
        if (!position.updateById()) {
            throw new MqNotAckException("修改仓位记录失败");
        }
    }

    /**
     * 同步止盈止损订单
     *
     * @param dto dto
     * @since 2023/2/8 14:00
     */
    @Override
    public void sync(StrategySyncStopOrderMsgDTO dto) {
        try {
            initTradeClient(dto.getUserId());
        } catch (ApiRequestException e) {
            throw new MqNotAckException(e);
        }

        // 查询历史止盈止损
        String orderType = ApiOkxEnum.AlgoOrderType.getBySideType(dto.getSideType());
        JSONObject res = TRADE_CLIENT.get().getAlgoOrderHistory(null, dto.getExternalId().toString(), null, null, orderType, null, null, null);
        // 检查返回结果
        JSONObject stopOrderInfo = checkStopOrderDetail(dto, res);

        SyncStopOrderBO bo = new SyncStopOrderBO();
        bo.setMsg(dto);
        // 查询任务缓存
        bo.setTaskCache(strategyContractUtil.getStrategyTaskCache(dto.getTaskId(), false));
        // 开启本地事务
        mysqlUtil.startLocalTx();
        // 查询任务，并锁定
        QueryWrapper<StrategyTaskContract> queryTask = new QueryWrapper<>();
        queryTask.lambda()
                .ge(StrategyTaskContract::getSyncStopOrderNum, 1)
                .eq(StrategyTaskContract::getId, dto.getTaskId())
                .in(StrategyTaskContract::getStatus,
                        StrategyTaskContractEnum.Status.PAUSE.getValue(),
                        StrategyTaskContractEnum.Status.RUNNING.getValue())
                .last("for update");
        StrategyTaskContract task = strategyTaskContractMapper.selectOne(queryTask);
        if (Objects.isNull(task)) {
            throw new MqNotAckException("未能查询到运行中的任务信息");
        }
        bo.setTask(task);

        // 同步止盈止损信息
        sync(bo, stopOrderInfo);
        // 处理任务信息
        syncHandleTask(bo);
        // 处理仓位信息
        syncHandlePosition(bo);

        // 止盈止损成功触发
        if (bo.getIsSuccess()) {
            // 初始化订单信息，发送同步订单消息
            StrategyStopOrderContract stopOrder = bo.getStopOrder();
            PlaceOrderBO param = strategyTaskContractHandler.buildWhenStopOrderTrigger(stopOrder.getTaskId(),
                    stopOrder.getSide(),
                    stopOrder.getActualTriggerSide(),
                    stopOrder.getActualNum(),
                    stopOrder.getActualPrice(),
                    stopOrder.getPositionId());
            if (strategyOrderContractMapper.insert(param) != 1) {
                throw new MqNotAckException("新增订单信息失败");
            }

            // 缓存订单信息
            redisUtils.flushPersistentCache(RedisKeyEnum.STRATEGY_ORDER_CONTRACT, param.getId(), param);

            StrategySyncOrderMsgDTO<PlaceOrderBO> syncOrderMsg = new StrategySyncOrderMsgDTO<>();
            syncOrderMsg.setPlatformId(stopOrder.getPlatformId());
            syncOrderMsg.setOrderId(param.getId());
            syncOrderMsg.setExternalOrderId(stopOrder.getOrderId());
            syncOrderMsg.setSwitchId(dto.getSwitchId());
            syncOrderMsg.setIsStopOrder(true);
            bo.setSyncOrderMsg(syncOrderMsg);
        }

        // 修改缓存
        strategyContractUtil.flushTaskCacheByEntity(bo.getTaskCache(), bo.getTask(), null, null);
        // 提交本地事务
        mysqlUtil.commitLocalTx();

        // 发送同步订单消息
        if (bo.getIsSuccess()) {
            strategyContractUtil.sendMqSyncOrder(bo.getSyncOrderMsg());
        }
    }

    /**
     * 同步止盈止损-处理仓位
     *
     * @param bo bo
     * @since 2023/2/9 20:05
     */
    private void syncHandlePosition(SyncStopOrderBO bo) {
        StrategyStopOrderContract stopOrder = bo.getStopOrder();
        StrategyTaskContractCacheBO taskCache = bo.getTaskCache();

        // 查询两个仓位，并锁定
        List<StrategyPositionContract> positionList = strategyContractUtil.getPositionList(stopOrder.getTaskId());
        taskCache.setPositionList(positionList);

        // 根据方向获取订单对应的仓位
        StrategyPositionContract position = StrategyContractUtil.getCurrentPositionBySide(positionList, stopOrder.getSide(), false);
        if (position.getNumStopOrder() < stopOrder.getNum()) {
            String errorMsg = String.format("仓位止盈止损数量异常，仓位数量：%s，止盈止损委托数量：%s",
                    position.getNumStopOrder(),
                    stopOrder.getNum());
            throw new MqNotAckException(errorMsg);
        }

        position.setNumStopOrder(position.getNumStopOrder() - stopOrder.getNum());
        position.setUpdateTime(LocalDateTime.now());
        position.setUpdateBy(0L);
        if (!position.updateById()) {
            throw new MqNotAckException("修改仓位状态失败");
        }
    }

    /**
     * 同步止盈止损-处理任务
     *
     * @param bo bo
     * @since 2023/2/9 19:58
     */
    private void syncHandleTask(SyncStopOrderBO bo) {
        StrategyTaskContract task = bo.getTask();
        StrategyTaskContractCacheBO taskCache = bo.getTaskCache();

        task.setUpdateTime(LocalDateTime.now());
        task.setUpdateBy(0L);
        task.setUpdateName("系统");
        task.setSyncStopOrderNum(task.getSyncStopOrderNum() - 1);
        // 如果止盈止损成功
        if (bo.getIsSuccess()) {
            task.setPendingOrderNum(task.getPendingOrderNum() + 1);
        }
        if (!task.updateById()) {
            throw new MqNotAckException("修改任务状态失败");
        }

        // 将完成的止盈止损从缓存中去除
        taskCache.removeStopOrderCache(bo.getStopOrder().getId());
    }

    /**
     * 同步止盈止损
     *
     * @param bo            bo
     * @param stopOrderInfo 止盈止损
     * @since 2023/2/9 16:08
     */
    public void sync(SyncStopOrderBO bo, JSONObject stopOrderInfo) {
        StrategySyncStopOrderMsgDTO dto = bo.getMsg();

        String state = stopOrderInfo.getString("state");
        boolean isSuccess = ApiOkxEnum.AlgoOrderState.EFFECTIVE.getName().equals(state);
        bo.setIsSuccess(isSuccess);

        // 查询止盈止损并锁定
        QueryWrapper<StrategyStopOrderContract> queryStopOrder = new QueryWrapper<>();
        queryStopOrder.lambda()
                .eq(StrategyStopOrderContract::getId, dto.getStopOrderId())
                .eq(StrategyStopOrderContract::getSyncStatus, StrategyStopOrderContractEnum.SyncStatus.RUNNING.getValue())
                .eq(StrategyStopOrderContract::getStatus, StrategyStopOrderContractEnum.Status.NOT_ACTIVE.getValue())
                .last("for update");
        StrategyStopOrderContract stopOrder = strategyStopOrderContractMapper.selectOne(queryStopOrder);
        if (Objects.isNull(stopOrder)) {
            throw new MqNotAckException("同步止盈止损时，未能查询到止盈止损信息，id：" + dto.getStopOrderId());
        }
        bo.setStopOrder(stopOrder);

        // 止盈止损成功触发，根据返回的订单信息修改止盈止损数据
        if (isSuccess) {
            // 止盈止损生成的订单的id
            stopOrder.setOrderId(stopOrderInfo.getLong("ordId"));

            // 实际触发方向
            String actualSide = stopOrderInfo.getString("actualSide");
            if (ApiOkxEnum.AlgoOrderActualSide.LOST.getName().equals(actualSide)) {
                stopOrder.setActualTriggerSide(StrategyStopOrderContractEnum.ActualTriggerSide.LOST.getValue());
            } else {
                stopOrder.setActualTriggerSide(StrategyStopOrderContractEnum.ActualTriggerSide.PROFIT.getValue());
            }

            // 触发时间
            Long triggerTimeMillis = stopOrderInfo.getLong("triggerTime");
            stopOrder.setTriggerTime(CommonUtil.millis2LocalDateTime(triggerTimeMillis));
            stopOrder.setActualNum(stopOrderInfo.getInteger("actualSz"));
            stopOrder.setActualPrice(stopOrderInfo.getBigDecimal("actualPx"));
        }

        // 设置状态
        stopOrder.setSyncStatus(StrategyStopOrderContractEnum.SyncStatus.DONE.getValue());
        stopOrder.setStatus(transferStopOrderStatus(state));
        if (strategyStopOrderContractMapper.updateById(stopOrder) != 1) {
            throw new MqNotAckException("修改止盈止损失败");
        }
    }

    /**
     * 转换ok止盈止损委托状态
     *
     * @param state ok止盈止损委托状态
     * @return {@link String}
     * @since 2023/2/9 17:01
     */
    private String transferStopOrderStatus(String state) {
        // 已生效
        if (ApiOkxEnum.AlgoOrderState.EFFECTIVE.getName().equals(state)) {
            return StrategyStopOrderContractEnum.Status.ACTIVATED.getValue();

            // 已撤销
        } else if (ApiOkxEnum.AlgoOrderState.CANCELED.getName().equals(state)) {
            return StrategyStopOrderContractEnum.Status.REVOKED.getValue();

            // 委托失败
        } else if (ApiOkxEnum.AlgoOrderState.ORDER_FAILED.getName().equals(state)) {
            return StrategyStopOrderContractEnum.Status.FAIL.getValue();

        } else {
            throw new MqNotAckException("ok同步止盈止损委托时，意外的状态值：" + state);
        }
    }

    /**
     * 检查止盈止损返回结果
     *
     * @param dto dto
     * @param res 返回结果
     * @return {@link JSONObject} 止盈止损
     * @since 2022/12/6 21:47
     */
    private JSONObject checkStopOrderDetail(StrategySyncStopOrderMsgDTO dto,
                                            JSONObject res) {

        boolean success = ApiOkxUtil.isSuccess(res, false);
        if (!success) {
            if ("Order does not exist".equals(ApiOkxUtil.getResMsg(res))) {
                throw new MqNotAckException("没有查询到ok应该触发的止盈止损");
            }

            // 设置止盈止损异常
            String reason = new StringBuilder("OK接口异常，返回结果：")
                    .append(res.toJSONString())
                    .toString();
            strategyTaskContractHandler.errorStopOrder(dto, reason, null);

            throw new MqAckException();
        }

        JSONArray data = res.getJSONArray(ApiOkxConstant.DATA);
        if (data.size() != 1 || Objects.isNull(data.getJSONObject(0))) {
            // 数据为空时，可能ok侧止盈止损还有来得及生效，根据策略上配置的重试次数进行重试
            StrategyContract strategy = strategyContractUtil.getStrategyCache(dto.getStrategyId(), false);
            // 累计重试次数
            Long count = incrSyncStopOrderCount(dto.getStopOrderId());
            if (count > strategy.getOrderRetryMax()) {
                // 设置止盈止损异常
                String reason = "重试多次后，未能查询到应被触发的止盈止损数据";
                strategyTaskContractHandler.errorStopOrder(dto, reason, --count);
            } else {
                // 重新发送同步信息
                strategyContractUtil.sendMqSyncStopOrder(dto);
            }

            throw new MqAckException();
        }

        return data.getJSONObject(0);
    }

    /**
     * 批量撤销止盈止损
     *
     * @param dto dto
     * @since 2023/2/10 15:10
     */
    @Override
    public void cancelBatch(StrategyStopOrderContractCancelBatchDTO dto) {
        try {
            initTradeClient(dto.getUserId());
        } catch (ApiRequestException e) {
            throw new MqNotAckException(e);
        }

        StrategyTaskContractCacheBO taskCache = strategyContractUtil.getStrategyTaskCache(dto.getTaskId(), false);
        if (!Objects.equals(StrategyTaskContractEnum.Status.PAUSE.getValue(), taskCache.getStatus())
                && !Objects.equals(StrategyTaskContractEnum.Status.RUNNING.getValue(), taskCache.getStatus())) {
            throw new MqAckException("撤销止盈止损时，任务状态不是暂停/运行中");
        }

        StrategyContract strategy = strategyContractUtil.getStrategyCache(taskCache.getStrategyId(), false);
        SystemProduct product = strategyContractUtil.getProductById(strategy.getProductId(), false);

        // 筛选批量需要批量撤销的止盈止损中除于未同步状态的数据
        List<StrategyStopOrderContractCacheBO> needCancelList = taskCache.getStopOrderList().stream()
                .filter(stopOrder -> Objects.equals(StrategyStopOrderContractEnum.SyncStatus.NOT_SYNC.getValue(), stopOrder.getSyncStatus())
                        && dto.getIdList().contains(stopOrder.getId()))
                .collect(Collectors.toList());

        // 一次最多批量处理10条
        int maxSize = 10;
        int endIndex = 0;
        for (int i = endIndex; i < needCancelList.size(); i = endIndex) {
            endIndex = i + maxSize;
            if (endIndex > needCancelList.size()) {
                endIndex = needCancelList.size();
            }
            // 分批处理
            List<StrategyStopOrderContractCacheBO> partList = needCancelList.subList(i, endIndex);
            // 生成请求参数集合
            List<CancelAlgoOrder> paramList = partList.stream()
                    .map(stopOrder -> {
                        // 设置缓存止盈止损同步状态
                        stopOrder.setSyncStatus(StrategyStopOrderContractEnum.SyncStatus.RUNNING.getValue());

                        CancelAlgoOrder param = new CancelAlgoOrder();
                        param.setInstId(product.getName());
                        param.setAlgoId(stopOrder.getExternalId().toString());
                        return param;
                    }).collect(Collectors.toList());

            // 开启本地事务
            mysqlUtil.startLocalTx();
            // 修改任务数据
            UpdateWrapper<StrategyTaskContract> updateTask = new UpdateWrapper<>();
            updateTask.lambda()
                    .setSql("sync_stop_order_num = sync_stop_order_num + " + partList.size())
                    .eq(StrategyTaskContract::getId, taskCache.getId());
            if (strategyTaskContractMapper.update(null, updateTask) != 1) {
                throw new MqNotAckException("批量撤销止盈止损时，修改任务状态失败");
            }

            // 收集id
            List<Long> idList = partList.stream()
                    .map(StrategyStopOrderContractCacheBO::getId)
                    .collect(Collectors.toList());
            // 修改止盈止损数据
            UpdateWrapper<StrategyStopOrderContract> updateStopOrder = new UpdateWrapper<>();
            updateStopOrder.lambda()
                    .set(StrategyStopOrderContract::getSyncStatus, StrategyStopOrderContractEnum.SyncStatus.RUNNING.getValue())
                    .eq(StrategyStopOrderContract::getSyncStatus, StrategyStopOrderContractEnum.SyncStatus.NOT_SYNC.getValue())
                    .eq(StrategyStopOrderContract::getStatus, StrategyStopOrderContractEnum.Status.NOT_ACTIVE.getValue())
                    .in(StrategyStopOrderContract::getId, idList);
            if (strategyStopOrderContractMapper.update(null, updateStopOrder) != idList.size()) {
                throw new MqNotAckException("批量撤销止盈止损时，修改止盈止损状态失败");
            }

            // 批量撤销
            try {
                JSONObject res = TRADE_CLIENT.get().cancelAlgoOrder(paramList);
                boolean success = ApiOkxUtil.isSuccess(res, false);
                if (!success) {
                    throw new MqAckLogException("ok接口异常，报错信息：" + ApiOkxUtil.getResMsg(res));
                }

            } catch (Throwable e) {
                /*
                 批量撤销的止盈止损中有不是待生效状态的时候，会报以下错误
                 ok接口异常，报错信息：Cancellation failed as the order doesn't exist
                 以下场景可能发生：
                 在此方法运行途中，交易所侧触发了某个止盈止损
                 */
                if (!e.getMessage().contains("Cancellation failed as the order doesn't exist")) {
                    throw e;
                }
            }

            // 修改任务缓存
            taskCache.setSyncStopOrderNum(taskCache.getSyncStopOrderNum() + partList.size());
            // 刷新任务缓存
            redisUtils.flushPersistentCache(RedisKeyEnum.STRATEGY_TASK_CONTRACT, taskCache.getId(), taskCache);

            // 提交本地事务
            mysqlUtil.commitLocalTx();

            // 遍历发送mq，同步止盈止损信息
            partList.forEach(cache -> {
                StrategySyncStopOrderMsgDTO syncStopOrderMsgDTO = new StrategySyncStopOrderMsgDTO();
                syncStopOrderMsgDTO.setPlatformId(cache.getPlatformId());
                syncStopOrderMsgDTO.setExternalId(cache.getExternalId());
                syncStopOrderMsgDTO.setStopOrderId(cache.getId());
                syncStopOrderMsgDTO.setStrategyId(strategy.getId());
                syncStopOrderMsgDTO.setSwitchId(taskCache.getSwitchId());
                syncStopOrderMsgDTO.setTaskId(taskCache.getId());
                syncStopOrderMsgDTO.setUserId(taskCache.getUserId());
                syncStopOrderMsgDTO.setSideType(cache.getSideType());

                strategyContractUtil.sendMqSyncStopOrder(syncStopOrderMsgDTO);
            });
        }
    }
}
